package org.tigr.htc.server.pbs;

/*
This library is free software; you can redistribute it and/or
modify it under the terms of the GNU Lesser General Public
License as published by the Free Software Foundation; either
version 2.1 of the License, or (at your option) any later version.

This library is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
Lesser General Public License for more details.

You should have received a copy of the GNU Lesser General Public
License along with this library; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
*/

import java.io.File;
import java.io.FileOutputStream;
import java.io.FileReader;
import java.io.FileWriter;
import java.io.IOException;
import java.io.LineNumberReader;
import java.io.PrintWriter;
import java.text.DateFormat;
import java.text.SimpleDateFormat;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.log4j.Logger;
import org.tigr.antware.shared.command.events.CommandInterrupt;
import org.tigr.antware.shared.util.RunSystemCommand;
import org.tigr.antware.shared.util.StringUtils;
import org.tigr.htc.cmd.CmdState;
import org.tigr.htc.cmd.Command;
import org.tigr.htc.cmd.CommandConfig;
import org.tigr.htc.cmd.CommandType;
import org.tigr.htc.cmd.ICommand;
import org.tigr.htc.cmd.Param;
import org.tigr.htc.cmd.ParamType;
import org.tigr.htc.common.HTCConfig;
import org.tigr.htc.server.IRunner;
import org.tigr.htc.server.RunnerMonitor;

/**
 * The <b>PBSRunner</b> class takes a command and creates a Sun Grid Engine job 
 * and launches the job on the grid.
 * <p>
 * The class is responsible for building the submit file based on the 
 * parameters of the command and submitting the job.  Then the runner 
 * registers itself with JobMonitor.
 */
public class PBSRunner implements IRunner {
    static Logger logger = Logger.getLogger(PBSRunner.class);
    
    /**
     * <code> cmd </code> the command to submit to the grid
     */
    private Command cmd;
    
    /**
     * <code> cwd </code> system working directory for this runner.
     * This is where the runner stores logs, submit files, etc.
     */
    private File cwd;
    
    /**
     * <code> CID </code> command id
     */
    private String gridID;
    
    /**
     * <code> eventReader </code> reader for the event log.
     */
    public LineNumberReader eventReader = null;
    
    /**
     * <code> eventLog </code> is the file where the grid will 
     * write events.
     */
    public File eventLog;
    
    private static String PBS_DIRECTIVE = HTCConfig.getProperty("pbs.directive");
    
    private static final String PRIORITY_GROUP = HTCConfig.getProperty("priority.group");
    
    private boolean arrayJob;
    
    // Job length variables
    private static final String FAST_JOB = "fast";
    private static final String MEDIUM_JOB = "medium";
    private static final String MARATHON_JOB = "marathon";
    
    // PBS Complexes 
    private static final String PRIORITY_COMPLEX = HTCConfig.getProperty("pbs.comp.priority");
    private static final String FAST_COMPLEX = HTCConfig.getProperty("pbs.comp.fast");
    private static final String MARATHON_COMPLEX = HTCConfig.getProperty("pbs.comp.marathon");
    
    // Checkpoint directory
    private static final String CKPT_DIR = HTCConfig.getProperty("pbs.ckpt.dir");
    
    
    // PBS Checkpoint
      private static final String PBS_KILL_TRAP_SCRIPT = HTCConfig.getProperty("pbs.kill.trap.script");
    
    // Complex String to be used when a job is recheduled due to
    // exceeding the resource limits
    private StringBuffer reschedComplexStr = new StringBuffer();
    
    private boolean eventReadersCreated = false;
    
    // Create hash for file handles of an array job's event log files
    
    HashMap<Integer, LineNumberReader> fileHash = null;
    
    // Variable that indicates if the current job is a marathon job
    boolean skipNotifyToQsub = false;
    
    
    /**
     * This constructor creates a new instance of PBS runner to launch the command. The
     * runner creates a submit file in a directory with the command ID under the specified
     * parent directory.
     *  
     * @param cmd a <code>Command</code> object that will executed by this runner
     * @param cwd a <code>File</code> representing the parent directory
     */
    public PBSRunner(Command cmd, File cwd) {
        logger.debug("PBS runner constructor using work dir " + cwd);
        this.cmd = cmd;
        this.cwd = new File(cwd, "" + cmd.getID());
        
        // Set the arrayJob flag
        arrayJob = (cmd.getTimes() > 0) ? true : false;
        
        // Set initial dir to 'cwd' if it is null
        logger.debug("Command's initial directory is " + cmd.getInitialDir());
        if ( (cmd.getInitialDir() == null) || cmd.getInitialDir().equals("/")) {
            cmd.setInitialDir(cwd.getAbsolutePath());
        }
    }
    
    /**
     * The method <code>createEventReaders</code> creates the event readers required to
     * monitor the job. If this is an array job then the array count number of readers
     * are created.
     * 
     * @throws IOException
     */
    public void createEventReaders() throws IOException {
        
        // Create an event log file for every job.
        eventLog = new File(cwd, "event.log");
        if (!eventLog.exists()) {
        	eventLog.createNewFile();
        }
        
        // Create array job event log files.
        if (arrayJob) {
            // Create the file hash
            fileHash = new HashMap<Integer, LineNumberReader> ();
            
            for (int i = 1; i <= cmd.getTimes(); i++) {
                logger.debug("Creating file event.log." + i + " in directory " + cwd);
                File fileEventLog = new File(cwd, "event.log." + i);
                if (!fileEventLog.exists()) fileEventLog.createNewFile();
                LineNumberReader eventReader = new LineNumberReader(
                        new FileReader(fileEventLog));
                fileHash.put(new Integer(i), eventReader);
            }
        } else {
            eventReader = new LineNumberReader(new FileReader(eventLog));
        }
        eventReadersCreated = true; // Set the flag which is used by monitoring method
    }

    /**
     * The method <code>buildArgs</code> builds a list of arguments from the specified 
     * command object. The command object has a list of <code>Param</code> objects which
     * may or may not contain key, but will always have values. So build the array from
     * these param objects.
     * 
     * @return a <code>String</code> array with the arguments to be passed to the command
     * invocation. 
     */
    private String [] buildArgs() {
        logger.debug("Building arguments array.");
        
        String [] args = new String [cmd.getParams().size()];
        Collection params = cmd.getParams();
        
        Iterator iter = params.iterator();
        int i = 0;
        for (; iter.hasNext(); i++) {
            Param param = (Param) iter.next();
            logger.debug("Processing parameter: " + param);
            
            // If this parameter has a key then specify as key value pair
            // otherwise just use the value as the argument
            if (param.getType() == ParamType.PARAM) {
                if (param.getKey() != null) {
                    args[i] = param.getKey() + "=" + param.getValue();
                } else {
                    args[i] = param.getValue();
                }
            } else {
                logger.warn("Illegal/unknown parameter type for command type htc.");
            }			
        }
        
        logger.debug("Arguments array: " + StringUtils.toString(args));
        return args;
    }
    
    /**
     * The method <code>run</code> builds a submit file for the command fires the submit
     * event and then submits the job to the underlying grid.
     * 
     */
    public void run() {
        PrintWriter pw = null;
        logger.debug("Submitting job represented by command: " + cmd);
        
        try {
            // setup make working directory. If one already exists then delete it 
        	// and recreate the directories as needed
            if (cwd.exists()) {
                logger.warn("CWD exists, going to delete ");
                File [] files = cwd.listFiles();
                if (files.length != 0) {
                    logger.warn("Directory is not empty. Deleting files first.");
                    for (int i = 0; i < files.length; i++) {
                        files[i].delete();
                    }
                }
                
                // After any files in the directory have been removed, remove the 
                // directory 
                if (!cwd.delete()) {
                    logger.error("Couldn't delete working directory needed by server " + cwd);
                }
            }
            
            if (!cwd.mkdirs()) {
                logger.error("Couldn't create working directory needed by server " + cwd);
                cmd.fireFailureEvent(new Exception(" Couldn't create working directory needed by server " + cwd));
                return;
            }
           
            // Create event log file for the array job
            try {
                createEventReaders();
            } catch (Exception e) {
                // TODO: handle exception
                logger.warn("Error creating the event log files for command " + cmd.getID() + " " + e.getMessage(), e);
                return;
            }
            
            // create writer for submit file
            File pbsSubmit = new File(cwd, "pbs_job.sh");
            if (logger.isDebugEnabled()) {
                logger.debug("Creating submit file: " + pbsSubmit.getAbsolutePath());
            }
            pw = new PrintWriter(new FileWriter(pbsSubmit), true);
            
            initTasks();
            
            // create submit file
            createSubmitFile(pw);
            
            // Submit the command
            gridID = PBSUtils.submit(cmd.getUserName(), cmd.getInitialDir(), cwd, skipNotifyToQsub);
            if (gridID == null) {
                logger.warn("Error running PBS submit. Fire failure notice!");
                cmd.fireFailureEvent(
                        new Exception("Command's grid ID was zero which is invalid"));
                return;
            }
                        
            // Set the grid job ID for the command and set and update the
            // state of the command in the database so status is reflected immediately
            cmd.setGridJobID(gridID);
            cmd.setState(CmdState.WAITING);
            cmd.updateCommand();
            
            // EVENT: Submit
            logger.info("Sending submit event");

            cmd.fireSubmitEvent(cmd.getSubmitURL(), eventLog.getAbsolutePath());
            
            // monitor log file
            logger.debug("Done submitting command with HTC ID: " + cmd.getID() + 
                    " cid is " + cmd.getGridJobID() + " now just wait in monitor mode");
            
            RunnerMonitor.register(this);
            
        } catch (Exception e) {
            String message = e.getMessage();
            logger.debug("Error submitting job to the grid " + message);
            
            // If submission failed, check if there is any PBS error message in 
            // pbs_submit.out file
            File pbsSubmitOut = new File(cwd, "pbs_submit.out");
            if (pbsSubmitOut.exists()) {
            	logger.debug("Reading " + pbsSubmitOut.getName() + " file for error.");
            	try {
					String line = null;
					LineNumberReader fileReader = new LineNumberReader(
							new FileReader(pbsSubmitOut));
					while ((line = fileReader.readLine()) != null) {
						message = message + "\n" + line;
					}
					fileReader.close();
				} catch (Exception lne) {
					// TODO: handle exception
				}
            }
            
            // Build a new exception with new error message.
            Exception newException = new Exception(message);
            logger.warn("PBS Runner error " + message, newException);
                  
            cmd.fireFailureEvent(newException);
            
            // Added this line to persist the 'FAILED' state to the command 
            // if any exceptions occur while submitting the jobs to the grid.
            cmd.updateCommand();
            
            return;
        } finally {
            // always close PBS submit file
            if (pw != null) {
                pw.close();
            }
        }
        
    }
    
    
    
    /**
     * The method <code>initTasks</code> initializes calculates the
     * the number of tasks this command contains.  For this runner
     * it just initializes to times.   
     * 
     */
    public void initTasks() {
        int tasks = cmd.getTimes();
        cmd.setNumTasks(tasks);
        cmd.getStatus().setWaiting(tasks);
    }
    

    private static String parseToken(Pattern pattern, String line) {
        String token = null;
        Matcher matcher = pattern.matcher(line);
        if (matcher.matches()) {
            // As the pattern matches extract the token of interest
            token = matcher.group(1);
        }
        
        return token;
    }
    /**
     * The method <code>recreateCommand</code> recreates the HTC Command object from the
     * request ID. This method parses the pbs_job.sh file to recreate the command is 
     * typically used when reconnecting to a previous request in a local HTC Service context.
     * 
     * @return a <code>Command</code> object representing this request.
     * @throws IOException 
     */
    public static Command recreateCommand(String workingDir, long htcID) throws IOException {
        logger.debug("Attempting to recreate command for htc ID: " + htcID + 
                " with working dir '" + workingDir);
        Command cmd = new Command();
        cmd.setType(CommandType.PBS);
        cmd.setID(htcID);
        CommandConfig config = new CommandConfig();
        
        
        // Setup the regular expression patterns
        Pattern outputPattern = Pattern.compile(PBS_DIRECTIVE + " -o (.+)");
        Pattern errorPattern = Pattern.compile(PBS_DIRECTIVE + " -e (.+)");
        Pattern timesPattern = Pattern.compile(PBS_DIRECTIVE + " -t 1-(\\d+)");
        Pattern runTimePattern = Pattern.compile(PBS_DIRECTIVE + " -l h_rt=(\\w+)");
        Pattern fastComplexPattern = Pattern.compile(PBS_DIRECTIVE + " -l " + PBSRunner.FAST_COMPLEX);
        Pattern marathonComplexPattern = Pattern.compile(PBS_DIRECTIVE + " -l " + PBSRunner.MARATHON_COMPLEX);
        Pattern hostnamePattern = Pattern.compile(PBS_DIRECTIVE + " -l host=(.+)");
        Pattern archPattern = Pattern.compile(PBS_DIRECTIVE + " -l arch=(.+)");
        Pattern memoryPattern = Pattern.compile(PBS_DIRECTIVE + " -l mem=(\\d+)M");
        Pattern priorityPattern = Pattern.compile(PBS_DIRECTIVE + " -p (\\w+)");
        Pattern groupPattern = Pattern.compile(PBS_DIRECTIVE + " -A (\\w+)");
        Pattern evictPattern = Pattern.compile(PBS_DIRECTIVE + " -r (\\w)");
        Pattern passthroughPattern = Pattern.compile("# passthrough given !!!");
        Pattern cmdPattern = Pattern.compile("# Command: (.+)");
        Pattern argsPattern = Pattern.compile("# Args: (.+)");
        Pattern envPattern = Pattern.compile("# ENVIRONMENT: (\\w+)");
        Pattern cwdPattern = Pattern.compile("# cd (.+)");
        
        // create reader for submit file     
        File pbsSubmit = new File(workingDir + "/" + htcID, "pbs_job.sh");
        LineNumberReader lnr = new LineNumberReader(new FileReader(pbsSubmit));
        String line = null;
        boolean passthrough = false;
        
        while ((line = lnr.readLine()) != null) {
        	
            String token = null;
            if (logger.isDebugEnabled()) {
                logger.debug("Processing line: " + line);
            }
            // If the passthrough flag is true then this line should be the passthrough 
            // string so read it and reset the flag
            if (passthrough) {
                config.setPassThrough(line.substring(PBS_DIRECTIVE.length()));
                passthrough = false;
            }
            
            // Try to match output pattern
            if ((token = parseToken(outputPattern, line)) != null) {
                cmd.setOutput(token);
            }
            else if ((token = parseToken(envPattern, line)) != null) {
            	// If user has specified any environment, set the getenv flag
            	// to true.
                logger.debug("Found environment pattern: Token is " + token);
                
                if (token.equals("USER")){
                	logger.debug("Found user environment pattern: ");
                	config.setGetenv(true);
                }
            	else if (token.equals("DEFAULT")){
            		// If user has not specified any environment, set the getenv flag
            		// to false, so that the default environment is used
            		logger.debug("Found Default Environment Pattern: ");
            		config.setGetenv(false);
            	}
            }
            else if ((token = parseToken(errorPattern, line)) != null) {
            	// Try to match error pattern
                cmd.setError(token);
            }
            // Try to match error pattern
            else if ((token = parseToken(timesPattern, line)) != null) {
                int times = Integer.parseInt(token);
                cmd.setTimes(times);
            }
            // Try to match running time pattern
            else if ((token = parseToken(runTimePattern, line)) != null) {
                String [] timeParts = token.split(":");
                long runningTime = Long.parseLong(timeParts[0]) * 60 + 
                    Long.parseLong(timeParts[1]);
                config.setRunningTime(runningTime);
            }
            // Try matching job duration to fast or marathon
            else if (fastComplexPattern.matcher(line).matches()) {
                config.setLength(PBSRunner.FAST_JOB);
			} else if (marathonComplexPattern.matcher(line).matches()) {
                config.setLength(PBSRunner.MARATHON_JOB);
            } else if ((token = parseToken(hostnamePattern, line)) != null) {
            	// Try to match running hostname pattern
                config.setHosts(token);
            }
            // Try to match architecture pattern
            else if ((token = parseToken(archPattern, line)) != null) {
            	String [] archs = token.split("\\|");
            	String [] realArchs = new String [archs.length];

            	logger.debug("Arch array: " + StringUtils.buildArrayString(archs));

            	// Get the right architecture strings from configuration file
            	for (int i=0; i<archs.length; i++) {
            		realArchs[i] = HTCConfig.getProperty(archs[i]);
            	}
            	
            	logger.debug("Real arch array: " + StringUtils.buildArrayString(realArchs));
                config.setOpSys(StringUtils.buildArrayString(realArchs));
            } else if ((token = parseToken(memoryPattern, line)) != null) {
            	// Try to match memory pattern
                config.setMemory(Integer.parseInt(token));
            } else if ((token = parseToken(priorityPattern, line)) != null) {
            	// Try to match priority pattern
                config.setPriority(Integer.parseInt(token));
            } else if ((token = parseToken(groupPattern, line)) != null) {
            	// Try to match group pattern
            	cmd.setGroup(token);
            }
            // Try to match evict pattern
            else if ((token = parseToken(evictPattern, line)) != null) {
                if (token.equalsIgnoreCase("y")) {
                    config.setEvictable(true);
                }
                else {
                    config.setEvictable(false);
                }
            }
            // If the passthrough comment matches then set the passthrough flag
            else if (passthroughPattern.matcher(line).matches()) {
                passthrough = true;
            }
            // Try to match command pattern
            else if ((token = parseToken(cmdPattern, line)) != null) {
                logger.debug("Found Cmd pattern: ");
                cmd.setCommand(token);
            }
            // Try to match command's initial directory pattern
            else if ((token = parseToken(cwdPattern, line)) != null) {
                logger.debug("Found cwd pattern: Token is " +token);
                cmd.setInitialDir(token);
            }
            // Try to match args pattern
            else if ((token = parseToken(argsPattern, line)) != null) {
                // TODO: The way this code is written if one of the arguments had a = then
                // this will be split into key value, make sure that is not a problem.
                String [] args = token.split("\t"); // Split the arguments on tabs to get individual args
                for (int i = 0; i < args.length; i++) {
                    String arg = args[i];
                    int equalsIndex = arg.indexOf("=");
                    if (equalsIndex != -1) {
                        cmd.addParam(arg.substring(0, equalsIndex), arg.substring(equalsIndex + 1));
                    }
                    else {
                        cmd.addParam(arg);
                    }
                }
                //cmd.setCommand(token);
            }
        }
        lnr.close();
        
        // Once the command has been recreated read the event file to determine the current
        // state of the command
        logger.debug("Recreated command: " + cmd);
        cmd.setConfig(config);
        return cmd;
    }
    
    /**
     * The method <code>createSubmitFile</code> creates a submit file that can be used
     * to execute the command on the GRID
     * 
     * @param out a <code>PrintWriter</code> representing the submit file 
     * <i>pbs.submit</i> for this command.
     */
    private void createSubmitFile(PrintWriter out) {
        logger.debug("In create submit file ");
		
		// Check to see if this request is for an array job in which case set the array 
		// flag. This is done by checking the times specified in the command
        logger.debug("Times is set to " + cmd.getTimes() + " for Command " + cmd.getID());
		
		
        out.println("#!/bin/bash");
        out.println("#");
        out.println("# generated PBS submit file for HTC command " + cmd.getID());
        out.println("# submitted for user " + cmd.getUserName());
        out.println("# There are four sections to this file");
        out.println("# 1. PBS Configuration settings");
        out.println("# 2. Environment settings");
        out.println("# 3. Running outside of PBS");
        out.println("# 4. Run Command");
        
        out.println("############");
        out.println("# Section 1: PBS Configuration");
        out.println("############");
        out.println("");
        out.println("# PBS Interpret this script with the Bourne shell");
        out.println(PBS_DIRECTIVE + " -S /usr/local/bin/bash");
        out.println("");
        
        
        out.println("# NOTE: you need to 'cd' to the intial dir before submitting");
        out.println(PBS_DIRECTIVE + " -cwd");
        
        out.println("");
        
        // See if the user has specified redirection for STDIN and STDOUT  
        // in which case set those
        String output = cmd.getOutput();
        logger.debug("Command output directory is : " + output);
        
        String error = cmd.getError();
        logger.debug("Command error directory is : " + error);
        
        String input = cmd.getInput();
        logger.debug("Command error directory is : " + input);
        
        // If this is an array job see if the output and error specification have
		// to have their index token replaced with the specific PBS task id specified
		// as the environment variable by the N1 grid engine
		if (arrayJob) {
		    if (output != null) {
		        output = output.replaceAll("\\$\\(Index\\)", "\\$PBS_TASK_ID");
		    }
		    
		    if (error != null) {
		        error = error.replaceAll("\\$\\(Index\\)", "\\$PBS_TASK_ID");
		    }		    	
		    
		    if (input != null) {
		        input = input.replaceAll("\\$\\(Index\\)", "\\$PBS_TASK_ID");
		    }		    	
		}

        if (output != null) {
            out.println("# STDOUT goes to ");
            out.println(PBS_DIRECTIVE + " -o " + output);
        }
        
        if (error != null) {
            out.println("# STDERR goes to ");
            out.println(PBS_DIRECTIVE + " -e " + error);
        }
        if (error != null && output != null && output.equals(error)) {
            // As the output and error are same set the -j flag to indicate
            out.println(PBS_DIRECTIVE + " -j y");
        }
        out.println("");
        
        
		if (arrayJob) {
			out.println(PBS_DIRECTIVE + " -t 1-" + cmd.getTimes() + ":1"); // Set the array specification
		}

        out.println();
        out.println("# config variables");
        
        CommandConfig config = cmd.getConfig();
        if (config != null) {
        	// Check to see if the user has specified any running time for this 
        	// request
        	String runningTime = null;
        	if (config.getRunningTime() > 0) {
        		// If a running time has been specified, then convert that 
        		// into appropriate string
        		runningTime = PBSUtils.getTimeinHHmmSS(config.getRunningTime());
        	}

            // If the user has specified duration then use that to determine the 
            // complex. Presently the duration is ignored as no complexes exists 
            // for duration
        	if(config.getLength() != null) {
        		// Get length.
        		String length = config.getLength();
        		logger.debug("Length of the jobs is " + length);
        		
        		// 'length' could be 'fast'/'short', or 'medium', or 'long' 
        		// or 'indefinite'.
        		if (length.equals(PBSRunner.FAST_JOB)) {
        			// This job is marked as a short job or fast, so request
        			// 'fast' complex while submitting job
                	out.println(PBS_DIRECTIVE + " -l " + PBSRunner.FAST_COMPLEX);
        		} else if(length.equals(PBSRunner.MEDIUM_JOB)) {
        			// The default length of the jobs would be medium, set in CommandConfig
        			// class.
        			
        			// This is marked as a medium length, so no need to request
        			// any complexes. By default, these jobs will go to default.q 
        		} else if (length.equals(PBSRunner.MARATHON_JOB)) {
					// This job is marked as a long or indefinite job
        			// So, request marathon complex
        			out.println(PBS_DIRECTIVE + " -l " + PBSRunner.MARATHON_COMPLEX);
        			
        			// Set the variable that indicates that this job is a marathon job
        			skipNotifyToQsub = true;	
        		} else {
        			// No need to request any complexes here. These jobs will
        			// go to the deafault.q
        		}
        		
        	}
        	
        	// Build the running time into the submit options
        	if (runningTime != null) {
        		out.println(PBS_DIRECTIVE + " -l h_rt=" + runningTime);
        	}
        	
            // If the user has specified a passthrough then set that. This is done first
            // so that if the user explicitly overrode the passthrough they will be
            // honored
            if (config.getPassThrough() != null && ! config.getPassThrough().equals("")) {
            	logger.debug("PassThorugh String is specified : " +config.getPassThrough());
                out.println("# passthrough given !!!");
                out.println(PBS_DIRECTIVE + " " + config.getPassThrough());
                out.println("");
            }
            
            // if there are any hosts then set the hosts requirement
            if (config.getHosts() != null) {
                out.println("# hostname was given as a requirement");
                out.println(PBS_DIRECTIVE + " -l host=" + config.getHosts());
                out.println("");
                // Add it to the reschedule complex string 
                reschedComplexStr.append(" -l host=" + config.getHosts());
            }
            
            // If the operating system or architecture is specified set that directive
            if (config.getOpSys() != null) {
                // Make sure that the user has specified a valid set of parameters
                logger.debug("Opsys String is " + config.getOpSys());
                String[] ops = config.getOpSys().split(",");
                logger.debug("Number of operating systems specified is " + ops.length);
                
                // As the -l option seems to not be additive if they are on different lines
                // make sure that the architecture string is comma separated
                //StringBuffer sbuf = new StringBuffer(" -l arch=");
                StringBuffer sbuf = new StringBuffer();
                
                for (int i = 0; i < ops.length; i++) {
                    if (i > 0) {
                        sbuf.append("|"); // Append the comma if this is not the first entry
                    }
                    logger.debug("Opsys is " + ops[i]);
                    String arch = HTCConfig.getProperty(ops[i].trim().toLowerCase());
                    logger.debug("Architecture string for " + ops[i] + " is " + arch);
                    sbuf.append(arch);
                }
                
                if (sbuf.length() != 0) {
                	out.println(PBS_DIRECTIVE + " -l arch=" + sbuf.toString());
                	out.println("");
                	// Add it to the reschedule complex string 
                	reschedComplexStr.append(" -l arch=" + sbuf.toString());
                }
            }
            
            logger.debug(" Setting Memory requirements");
            
            // If the memory requirements are specified set them, append the 'M'
            // character as the memory requirements are specified in megabytes without the
            // M in the workflow while the sun grid specification requires the 'M'
            if (config.getMemory() != 0) {
                out.println(PBS_DIRECTIVE + " -l mem=" + config.getMemory() + "M");
                // Add it to the reschedule complex string 
                reschedComplexStr.append(" -l mem=" + config.getMemory() + "M");
            }
            
            // Set the user specified priority
            // TODO: priority should be change to queue
            if (config.getPriority() > 0) {
                out.println(PBS_DIRECTIVE + " -p " + config.getPriority());
            }
            
            // Use the group as the account string
            if (cmd.getGroup() != null) {
            	out.println(PBS_DIRECTIVE + " -A " + cmd.getGroup());
            }
            
            // Check to see if this job is evictable. If it is evictable 
            // build into submit string. Also, set rerunnable to 'true'
            // By requesting '-r y'
            if (config.isEvictable()) {
            	// Marathon jobs are always non evictable, so ignore this for 
            	// marathon jobs
            	if (! config.getLength().equals(PBSRunner.MARATHON_JOB)) {
            		out.println(PBS_DIRECTIVE + " -r y ");
            	}
            } else {
            	out.println(PBS_DIRECTIVE + " -r n ");
            	skipNotifyToQsub = true;
            }
            
            // getCmdClass method from config should give a project or team name
            logger.debug("Project name specified is :" + cmd.getConfig().getCmdClass() + ":");

            String team = cmd.getGroup();
            logger.debug("Project name specified is :" + team + ": and the length is " + team.length());
            if ( (team != null) && (team.length() > 0)) {
                logger.debug("Team name specified is : " + team);
                if (team.equalsIgnoreCase(PBSRunner.PRIORITY_GROUP)) {
                    out.println(PBS_DIRECTIVE + " -l " + PBSRunner.PRIORITY_COMPLEX);
                } else {
                	// There is no need to request a default queue name.
                }
            } else { 
            	// There is no need to request a default queue name.
            }
        }
        
        // Set the name of the command to the name of the actual command without path
        String cmdName = cmd.getCommand();
        int lastSepIndex = cmdName.lastIndexOf(File.separatorChar);
        cmdName = (lastSepIndex == -1) ? cmdName : cmdName.substring(lastSepIndex + 1);
        out.println(PBS_DIRECTIVE + " -N " + cmdName);
        out.println();
        
        out.println("############");
        out.println("# Section 2: Environment Setup Configuration");
        out.println("############");
        
        out.println("# ");
        out.println(PBS_DIRECTIVE + " -v htc_id=" + cmd.getID() + ",request_cwd=" + cwd.getAbsolutePath());
        out.println("");
        
        
        if (config.isGetenv()) {
            logger.debug("User has specified that the environment be propagated. So grab ENV.");
            
            if ((cmd.getEnv() != null) && (cmd.getEnv().size() > 0)) {
            	out.println("# ENVIRONMENT: USER");
		    	Map<String, String> envMap = cmd.getEnv();

		    	for (String envVar : envMap.keySet()) {
					// Do not pass the environment variable 'SGE_TASK_ID' as this causes
					// problems for jobs that were distributed by some other distributed 
					// jobs.
					if (! envVar.equals("PBS_TASK_ID")) {
						out.println("export " + envVar + "=\"" + envMap.get(envVar) + "\"");
					}
				}
            } else {
            	// Here PBS gets the environment from the submit machine, which is 
            	// one of the workers.
            	out.println("# ENVIRONMENT: DEFAULT");
                logger.debug("Let PBS get the server environment as the default.");
                out.println("# copy current environment");
                out.println(PBS_DIRECTIVE + " -V ");
            }
            out.println("");
        }
        
        out.println("############");
        out.println("# Section 3: Uncomment to run outside of pbs ");
        out.println("############");
        out.println("export htc_id=" + cmd.getID());
        out.println("export request_cwd=" + cwd.getAbsolutePath());
        out.println("export CKPT_DIR=" +CKPT_DIR);
        out.println("# cd " + cmd.getInitialDir());
        out.println("sigcounter=0");
        out.println("");
        
        out.println("############");
        out.println("# Section 4: Run Command");
        out.println("############");
        
        out.println("trap \"sigusr2 2\" USR2");
        out.println("trap \"sigusr1 1\" USR1");

        // Trap method for USR2
        out.println("sigusr2() {");
        out.println("  SIGNAL=$1");
        out.println("  echo \"I~~~`date` Job $JOB_ID caught signal $SIGNAL. \" >> ${request_cwd}/event.log");
        out.println("  MARATHON=false");
        out.println("  # Apparently, PBS notifies the script multiple times about the ");
        out.println("  # pending KILL signal, but only reschedule it on the 1st notification");
        out.println("  sigcounter=`echo $(($sigcounter + 1))`");
        out.println("  if [ \"$sigcounter\" -eq 1 ]; then");
        out.println("  	  echo \"I~~~`date` ~~~ ${htc_id}~~~Job $JOB_ID is being interrupted\" >> ${request_cwd}/event.log");
        out.println("       # sleeping for 10");
        out.println("       sleep 10");
        out.println("  	  echo \"I~~~`date` ~~~ ${htc_id}~~~Process ${user_cmd_proc_id} is interrupted.\" >> ${request_cwd}/event.log");
        out.println("  fi");
        out.println("}");

        // Trap method for USR1
        out.println("sigusr1() {");
        out.println("  SIGNAL=$1");
        out.println("  echo \"I~~~`date` Job $JOB_ID caught signal $SIGNAL. \" >> ${request_cwd}/event.log");
        out.println("  MARATHON=false");
        out.println("  # Apparently, PBS notifies the script multiple times about the ");
        out.println("  # pending KILL signal, but only reschedule it on the 1st notification");
        out.println("  sigcounter=`echo $(($sigcounter + 1))`");
        out.println("  if [ \"$sigcounter\" -eq 1 ]; then");
        out.println("  	echo \"I~~~`date` ~~~ ${htc_id}~~~Job $JOB_ID is being interrupted\" >> ${request_cwd}/event.log");
        out.println("     user_cmd_proc_id=`cat ${request_cwd}/pid.log`");
        out.println("  	echo \"I~~~`date` ~~~ ${htc_id}~~~ kill -2 -$user_cmd_proc_id\" >> ${request_cwd}/event.log");
        out.println("     kill -2 -$user_cmd_proc_pid");
        out.println("     # Call trap.sh, which alters the job and reschedules it");
        out.println(      PBS_KILL_TRAP_SCRIPT +  " $JOB_ID $SIGNAL $CKPT_DIR \"" + reschedComplexStr + "\"");
        out.println("     trap_ret=$?");
        out.println("     echo \"I~~~`date` Job $JOB_ID altered and rescheduled with return value $trap_ret\" >> ${request_cwd}/event.log");
        out.println("  	  echo \"I~~~`date` ~~~ ${htc_id}~~~Process ${user_cmd_proc_id} is interrupted.\" >> ${request_cwd}/event.log");
        out.println("  fi");
        out.println("}");
	    out.println("echo \"I~~~ wrapper script starting job \" >> ${request_cwd}/event.log");
        
        StringBuffer pbsCmd = new StringBuffer();
        out.println("# Command: " + cmd.getCommand());
        
        // If the number of times is set to be greater than zero then call the 
        // wrapper instead of the actual command so the wrapper can replace the token
        // Also tell PBS to launch an array
        if (arrayJob) {
            pbsCmd.append(cmd.getCommand() + " ");
        } else {
            pbsCmd.append(cmd.getCommand() + " ");
        }
        
		// Build the list of arguments based on the command parameters and do a token
		// replacement if this is an array job
        String [] args = buildArgs();
        for (int i = 0; i < args.length; i++) {
			if (arrayJob) {
				pbsCmd.append(" " + args[i].replaceAll("\\$\\(Index\\)", "\\$\\{PBS_TASK_ID\\}"));
			} else {
				pbsCmd.append(" " + args[i]);
			}
        }
        // Print the arguments as a tab separated list so these can be read when
        // recreating commands
        out.println("# Args: " + StringUtils.buildArrayString(args, "\t"));

        if (input != null) {
        	pbsCmd.append(" < " + input);
        }
        
        //Run command in background
        out.println(pbsCmd + " &");
        
        out.println("# Get the pid of the above job and set as an environemnt variable");
        out.println("user_cmd_proc_id=$!");
        out.println("echo \"$user_cmd_proc_id\" > ${request_cwd}/pid.log");
        out.println("echo \"I~~~ Job Process id is $user_cmd_proc_id \">> ${request_cwd}/event.log");
        
        out.println("# Wait for the above background job to finish"); 
        out.println("wait %%");
        
        out.println("pbs_ret_val=$?");
        
        out.println("sdate=$(date)");
        
        if (arrayJob) {
            // If this is an array job then print the task ID to the event log
            out.println("if [ -e \"${CKPT_DIR}/${JOB_ID}.${PBS_TASK_ID}.resched\" ]; then");
            out.println("    echo \"I~~~ Job Rescheduled \" >> ${request_cwd}/event.log.${PBS_TASK_ID}");
            out.println("    rm -f ${CKPT_DIR}/${JOB_ID}.${PBS_TASK_ID}.resched");
            out.println("else");
            out.println("     echo \"F~~~${htc_id}~~~${JOB_ID}.${PBS_TASK_ID}~~~$sdate~~~command finished~~~$pbs_ret_val\" >> ${request_cwd}/event.log.${PBS_TASK_ID}");
            out.println("fi");
        } else {
			// Print the job id only
            out.println("if [ -e \"${CKPT_DIR}/${JOB_ID}.resched\" ]; then");
            out.println("    echo \"I~~~ Job Rescheduled \">> ${request_cwd}/event.log");
            out.println("    rm -f ${CKPT_DIR}/${JOB_ID}.resched");
            out.println("else");
            out.println("    echo \"F~~~${htc_id}~~~${JOB_ID}~~~$sdate~~~command finished~~~$pbs_ret_val\" >> ${request_cwd}/event.log");
            out.println("fi");
        }
        
        out.println("# end of submit file ");
    }
    
    
    /* (non-Javadoc)
     * @see org.tigr.htc.server.IRunner#getICommand()
     */
    public ICommand getICommand() {
        return cmd;
    }
    
    /* (non-Javadoc)
     * @see org.tigr.htc.server.IRunner#stop()
     */
    public boolean stop() throws Exception {
    	logger.debug("Stopping grid job " + cmd.getGridJobID());
        PBSUtils.remove(cmd.getGridJobID());
        
        cmd.taskInterrupted("Command interrupted by user", CommandInterrupt.USER_KILL, 0);
        
        if (cmd.isFinished()) {
            if (cmd.getNotifyScript() != null) {
                runNotifyScript("Command interrupted by user", -1);
            }
        }
        
        // Some Commands are not marked interrupted, when issued a stop. 
        // request on them. Set the command state to interrupted explicitly 
        // here, if the state is not INTERRUPTED.
        if (! cmd.getState().equals(CmdState.INTERRUPTED)) {
        	cmd.setState(CmdState.INTERRUPTED);
        }
        
        return true;
    }
    
    /* (non-Javadoc)
     * @see org.tigr.htc.server.IRunner#hold()
     */
    public boolean hold() throws Exception {
        return PBSUtils.hold(gridID);
    }
    
    /* (non-Javadoc)
     * @see org.tigr.htc.server.IRunner#release()
     */
    public boolean release() throws Exception {
        return PBSUtils.release(gridID);
    }
    
    
    /* (non-Javadoc)
     * @see org.tigr.htc.server.IRunner#monitor()
     */
    public boolean isRunnerComplete() {
        // If this method is called when the readers not created it can cause a problem
        // so throw an exception.
        if (! eventReadersCreated) {
            throw new IllegalStateException("Attempt to call monitoring when readers have not been created.");
        }
        
		if (logger.isDebugEnabled()) {
			logger.debug(" monitor job " + cmd.getID());
			logger.debug(" file size " + eventLog.length());
		}
        
		if ((cmd.getState() == CmdState.FINISHED) || (cmd.getState() == CmdState.INTERRUPTED)) { 
        	logger.debug("Command is finished or interrupted, so don't bother checking files. Returning true.");
            return true;
        }
        
		// Check if the grid job of this command went into an 
		// error state for some reason. If it did, fire a 
		// task failure event

		// Check for 'Eqw' state using DRMAA API 
		int status = -1;
		
        try {

        	if (! arrayJob) {
                // make sure that the event reader is not null, this can happen when a job
                // is remonitored the event reader is not initialized
                if (eventReader == null) {
                    if (logger.isDebugEnabled()) {
                        logger.debug(" reopening event log file.");
                    }
                    eventReader = new LineNumberReader(new FileReader(eventLog));
                }

                // monitor event log file
                boolean complete = monitorForCompletion(eventReader);
               
                // Is success, close the event reader
                if (complete) {
                    eventReader.close();
                }
                
                return complete;
            } else {
                
                // Create Event log files, if the file hash is null. This could happen
                // in remonitoring the active jobs.
                if (fileHash == null) {
                    createEventReaders();
                }
                
                logger.debug("Monitor array job " +cmd.getID());
                // Monitor Array job files
                boolean arrayCompletion = true;
                
                Iterator itr = null;
                int i=1;
                
                for (itr = fileHash.keySet().iterator(); itr.hasNext(); i++) {
                    logger.debug("Monitoring the event log file event.log." +i);
                    // Get the line number reader and monitor it
                    Integer key = (Integer)itr.next();
                    LineNumberReader eventReader1 = (LineNumberReader)fileHash.get(key);
                    boolean indArrayCompletion = monitorForCompletion(eventReader1);
                    logger.debug("Monitoring event.log." + i + " is finished. Completion is " + indArrayCompletion);
                    if (indArrayCompletion) {
                    	logger.debug("Removing event reader from the iterator.");
                        itr.remove();
                        eventReader1.close();
                    }
                    
                    // Set the success of the array job
                    arrayCompletion = arrayCompletion && indArrayCompletion;
                }
                
                logger.debug(" Job with id "+ cmd.getID() + " is complete " + arrayCompletion);

                
                // If arrayCompletion is true, concatenate the individual event log file of the
                // array job.
                if (arrayCompletion) {
                    FileOutputStream fos = new FileOutputStream(eventLog);
                    String line = null;
                    
                    logger.debug("Merging the individual event log files for " + cmd.getID());
                    for (i=1; i <= cmd.getTimes(); i++ ) {
                        LineNumberReader fileReader = new LineNumberReader(
                                           new FileReader(new File(cwd, "event.log." +i)));
                        
                        while ( (line = fileReader.readLine()) != null) {
                            fos.write(line.getBytes());
                            fos.write("\n".getBytes());
                        }
                        
                        fileReader.close();
                        fos.flush();
                    }
                    logger.debug("Finished merging the individual event log files for " + cmd.getID());
                    fos.close();
                }
                
                return arrayCompletion;
            }
        } catch (Exception e) {
            logger.debug("Exception: " + e, e);
            return false;
        }
        
    }
    
    
    /**
     * The method <code>monitorForCompletion</code> method checks to see if this request
     * is omplete by monitoring the event.log file for event markers. The method then fires 
     * the appropriate events as needed.
     * 
     * @param eventReader
     * @return
     * @throws IOException a <code></code>
     */
    private boolean monitorForCompletion(LineNumberReader eventReader) throws IOException {
        String line;
        boolean returnValue = false;
        
        while ((line = eventReader.readLine()) != null) {
        	
            try {
                // parse line
                logger.debug("New event line.");
                if(line != null) {
                    String[] eventFields = line.split("~~~", 6);
                    
                    // event line must have at least a type and id
                    if(eventFields.length > 5) {
                        logger.debug("Event line " + line);
                        // parse line
                        char type = line.charAt(0);
                        int htc_id = Integer.parseInt(eventFields[1]);
                        int taskID = 0;
                        int pbs_id = 0;
                        // If this runner is for an array job then read the job ID and task ID
                        // else just read job ID
                        if (arrayJob) {
                            if (logger.isDebugEnabled()) {
                                logger.debug("Job ID field: " + eventFields[2]);
                            }
                            String [] idFields = eventFields[2].split("\\.");
                            pbs_id = Integer.parseInt(idFields[0]);
                            taskID = Integer.parseInt(idFields[1]);
                        }
                        else {
                            pbs_id = Integer.parseInt(eventFields[2]);
                        }
                        
                        String strDate = eventFields[3];
                        logger.debug("Event date from log '" + strDate+"'");
                        
                        String eventInfo = eventFields[4];
                        
                        SimpleDateFormat unixTimeStampFormatter = new SimpleDateFormat("EEE MMM dd hh:mm:ss z yyyy");
                        logger.debug("Current Date using unix formatter" + unixTimeStampFormatter.format(new Date()));
                        
                        Date eventDate = unixTimeStampFormatter.parse(strDate);
                        logger.debug("Date object after parsing using unix formatter " + eventDate);

                        DateFormat df = DateFormat.getDateTimeInstance(DateFormat.SHORT, DateFormat.SHORT);
                        
                        logger.debug("Date Time stamp after parsing using simple date formatter" + df.format(eventDate));
                        
                        switch (type) {
                            case 'S':  // start event
                                logger.debug("Start event with hostname " + eventFields[5]);
                                cmd.taskStarted(eventInfo, eventDate, eventFields[5], taskID, taskID);
                                break;
                            case 'F':  // finish event
                                int taskReturnValue = Integer.parseInt(eventFields[5].trim());
                                
                                cmd.taskFinished(eventInfo, eventDate, taskReturnValue, taskID);
                                if (cmd.isFinished()) {
                                    if (cmd.getNotifyScript() != null) {
                                        logger.debug("In sendNotification.");
                                        runNotifyScript(eventInfo, taskReturnValue);
                                    }
                                }
                                returnValue = true;
                                break;
                            case 'T': // info event
                                logger.debug("Interrupted or an unfinished event ");
                                break;
                            case 'I': // info event
                                logger.debug("Info event"); 
                                break;
                                
                            case 'E': // failure event
                                
                            default: //unknown event 
                                logger.debug("no event on line "); 
                            break;
                        }
                    } else {
                        if (logger.isDebugEnabled()) {
                            logger.debug("Line in event log that didn't match" + line);
                        }
                    }
                }
            } catch (Exception e) {
                logger.debug("Exception: " + e, e);
            }
        }
        
        logger.debug("Returning " + returnValue);
        return returnValue;
    }

    
    // method to run notify script and/or email user
    public void runNotifyScript(String message, int returnValue) {
        logger.debug("In runNotifyScript.");
        
        if (cmd.getNotifyScript() == null) {
            logger.warn("No script set.");
            return;
        }
        
        String email = cmd.getEmail();
        if (email == null) {
            email = cmd.getUserName();
        }
        
        File scriptFile = new File(cmd.getNotifyScript());
        String cid = "-requestID=" + cmd.getID();
        String returnval = "-returnValue=" + returnValue;
        String lmessage = "-message=" + message;
        String emailArg = "-email=" + email;
        
        if (scriptFile.exists()) {
            logger.debug("Notify script file exists");
            try {
                String tmp = scriptFile.getAbsolutePath();
                logger.debug("Notify cmd is " + tmp);
                logger.debug("Arguments: " + cid + returnval + lmessage + emailArg
                        + " -- ");

                RunSystemCommand run = new RunSystemCommand();
                String[] cmdsArray = new String[] { scriptFile.getAbsolutePath(), cid,
                        returnval, lmessage, emailArg };
                
                int exitVal = run.runCommand(cmdsArray);
                logger.debug("Process exit value: " + exitVal);
            } catch (Throwable e) {
            	logger.debug("Exception running notify scripts " + e.getMessage(), e);
			}
        } else {
            logger.debug("Notify script doesn't exist " + cmd.getNotifyScript());
        }
    }
}

